package com.yww.datastructure.blockingqueue;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author yww
 * @description 阻塞队列，单锁实现
 * @since 2024/2/7 12:02
 */
@SuppressWarnings("unchecked")
public class BlockingQueue1<E> implements BlockingQueue<E> {
    private final E[] array;
    private int size;
    private int head;
    private int tail;


    private final Lock tailLock = new ReentrantLock();  // 单锁
    private final Condition tailWaits = tailLock.newCondition();   // 生成线程等待队列
    private final Condition headWaits = tailLock.newCondition();  // 消费线程等待队列

    public BlockingQueue1(int capacity) {
        array = (E[]) new Object[capacity];
    }

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly(); // 获取锁
        try {
            while (isFull()) {
                tailWaits.await(); // 队列满，生产线程等待，临时释放锁
            }
            array[tail] = e;

            if (++tail == array.length) { // 让tail循环
                tail = 0;
            }
            size++;
            headWaits.signal(); // 唤醒一个消费线程
        } finally {

            tailLock.unlock();  // 解锁
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        try {
            // 将毫秒转化为纳秒
            long n = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()) {
                if (n <= 0) {
                    return false;  // 等待超过限制，返回false
                }
                n = tailWaits.awaitNanos(n); // 刷新剩余等待时间
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            size++;
            headWaits.signal(); // 唤醒一个消费线程
            return true;
        } finally {
            tailLock.unlock();
        }

    }

    @Override
    public E poll() throws InterruptedException {
        tailLock.lockInterruptibly(); // 获取锁
        try {
            while (isEmpty()) {
                headWaits.await(); // 队列空，消费者线程等待
            }
            E e = array[head];
            array[head] = null; // help gc
            if (++head == array.length) { // head 循环
                head = 0;
            }
            tailWaits.signal(); // 唤醒一个生产线程
            size--;
            return e;
        } finally {
            tailLock.unlock();
        }
    }

    @Override
    public E poll(long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        try {
            long n = TimeUnit.MILLISECONDS.toNanos(timeout); // 获取纳秒
            while (isEmpty()) {
                if (n <= 0) {
                    return null; // 超过等待限制，返回null
                }
                n = headWaits.awaitNanos(n); // 刷新剩余等待时间
            }
            E e = array[head];
            array[head] = null; // help gc
            if (++head == array.length) {
                head = 0;
            }
            tailWaits.signal(); // 唤醒一个生产线程
            size--;
            return e;
        } finally {
            tailLock.unlock();
        }
    }

    @Override
    public boolean isEmpty() {
        return size == 0;
    }

    @Override
    public boolean isFull() {
        return size == array.length;
    }

    @Override
    public int size() {
        return size;
    }
}
